<?php
class Server
{
	private $address = '0.0.0.0';
	private $port = '8081';
	private $ev;
	private $base;
	private $event_buffer_event;
	public 	$socket;
	private $clients = []; //客户端
	private $timeEvent = []; //计时器事件数组

	private $timer_is_open = 2;  //是否已经开启计时器 1是 2否
	
	function __construct()
	{
		$this->socket = stream_socket_server('tcp://'.$this->address.':'.$this->port, $errno, $errstr, STREAM_SERVER_BIND);
		//设置非阻塞
    	stream_set_blocking($this->socket, false);
		$cfg = new EventConfig();

		//禁止使用select事件方法
		if (!$cfg->avoidMethod("select") || !$cfg->avoidMethod("poll")) {
		    exit('设置失败');
		}

		//开启边缘触发
		// $cfg->requireFeatures($cfg::FEATURE_ET);
		// 创建与配置相关联的事件库
		$this->base = new EventBase($cfg);

		//监听新连接的到来然后触发回调
		$listener = new EventListener($this->base, array($this, "callback"), $this->socket, EventListener::OPT_REUSEABLE | EventListener::OPT_CLOSE_ON_FREE, -1, $this->socket);

		
		//设置事件侦听器的错误回调
		$listener->setErrorCallback(array($this, 'error'));

		if ($this->timer_is_open == 2) {
			$this->timer_is_open = 1;
			call_user_func(array($this, 'checkOnlineStatus'));
		}

		$this->base->loop();
		
	}
	//释放缓冲区
	public function __destruct ()
	{
		$this->event_buffer_event->free();
		foreach ($this->clients as &$c) $c = NULL;
	}

	
	/**
	 * 新连接来了的回调
	 * @param object $listener 事件监听对象
	 * @param int $fd 监听的文件描述符
	 * @param array $address 新连接的地址与所占端口
	 * @param $arg 用户给回调传的自定义数据(new EventListener(第三个参数就是传给回调的))
	 */
	public function callback ($listener,$fd,$address,$arg)
	{
		$base = $listener->getBase();
		
		//开启缓冲区事件
		$this->event_buffer_event = new EventBufferEvent($base,$fd, EventBufferEvent::OPT_CLOSE_ON_FREE);

		//设置缓冲区事件回调（读，写，状态更改）
		$this->event_buffer_event->setCallbacks(array($this, "read_callback"), NULL , array($this, "event_callback"), $fd);
		
		//启用对缓冲区事件的读取，写入或同时读取和写入事件
		$enable = $this->event_buffer_event->enable(Event::READ);
		if (!$enable) {
			$this->error('',1);
		}
		//开启低水位与高水位(设置读取的最小接收的字节数与最大可接收的字节数)
		$this->event_buffer_event->setWatermark(Event::READ, 3, 8192);
	}
	/**
	 * 读取缓冲区数据触发回调
	 * object $bev EventBufferEvent object
	 * $arg 通过EventBufferEvent :: __ construct（） 或 EventBufferEvent :: setCallbacks（）附加到所有回调的自定义变量
	 */
	public function read_callback ($bev, $fd)
	{
		if (!empty($this->clients)) {
			if (!array_key_exists((int)$fd, $this->clients)) {
				$header = $bev->read(1024);
				$this->handShake($fd, $header, $bev);
			}else {
				$con = $bev->read(8192);
				$con = $this->parse($con);
				if (!empty($con['type'])) {
					$type = $con['type'];
					switch ($type) {
						case 'online':	//上线
							$clients = $this->closeFd($this->clients, $con['user_id']);
							$this->online($clients, $con, $fd);
							//告诉前端开启心跳检测
							$this->openHeartbeats($this->clients[(int)$fd]['bev']);
							break;
						case 'broadcast'://广播，给全员发送通知
							# code...
							break;
						case 'chat'://聊天
							//private_chat-私聊  group_chat-群聊
							if ($con['chat_type'] == 'private_chat') {
								$to_uid = $con['to_uid'];
								foreach ($this->clients as $key => $value) {
									if ($value['user_id'] == $to_uid) {
										$arr = ['type' => 'chat', 'content' => $con['content'], 'message_id' => $con['message_id']];
										$this->onlySend($value['bev'],$arr);
									}
								}

							}else {
								$this->sendMessage($this->clients, $con, $fd);
							}
							break;
						case 'withdraw'://消息撤回
							call_user_func_array(array($this, 'messageWithdraw'),[$con['message_id'],$fd]);
							break;
						case 'heartbeats'://心跳检测
							//最后一次检测时间
							$this->clients[(int)$fd]['last_heartbeats_time'] = time();
							$this->onlySend($bev,['type' => 'heartbeats', 'content' => 'success']);
			    			break;
						default:
							
							break;
					}
				}
			}
		}else {
			$header = $bev->read(1024);
			$this->handShake($fd, $header, $bev);
		}
	}
	
	/**
	 * 消息撤回
	 *
	 */
	private function messageWithdraw ($message_id, $fd)
	{
		$con = json_encode(['type' => 'withdraw', 'message_id' => $message_id]);
		$con = $this->build($con);
		foreach ($this->clients as $key => $value) {
			if ((int)$key == (int)$fd) {
				continue;
			}
			$value['bev']->write($con);
		}
	}
	/**
	 * 关闭与当前缓冲区事件关联的文件描述符
	 * $clients 用户连接集合
	 * $user_id 正在与之通信的用户id
	 */
	private function closeFd ($clients, $user_id)
	{
		//因为每次客户端刷新页面会产生一个新的fd，所以将这个用户对应的旧的fd关闭并在数组中删除
		foreach ($clients as $key => $value) {
			if ($value['user_id'] == $user_id) {
				//关闭fd
				$value['bev']->close();
				unset($clients[$key]);
			}
		}
		return $clients;
	}

	/**
	 * 有新用户上线后给全员发送好友上线通知
	 * $clients 用户连接集合
	 * $con  正在与之通信的用户发送的上线通知内容集合
	 * $fd 正在与之通信用户的文件描述符
	 */
	private function online ($clients, $con, $fd)
	{
		$time = date("Y-m-d H:i:s");
		//发送上线通知
		foreach ($clients as $key => $value) {
			if ($clients[$key]['fd'] == $fd) {
				$clients[$key]['user_id'] = $con['user_id'];
				// $clients[$key]['online_time'] = time();
				$clients[$key]['last_heartbeats_time'] = time();
				continue;
			}
			$con['content'] = '用户'.$con['user_id'].' '.$time.'上线了';
			$data = $this->build(json_encode($con));
			$value['bev']->write($data);
		}
		$this->clients = $clients;
	}
	/**
	 * 聊天 - 发送消息
	 */
	private function sendMessage ($clients, $con, $fd)
	{
		foreach ($clients as $key => $value) {
			if ($clients[$key]['fd'] == $fd) {
				continue;
			}
			$data = $this->build(json_encode($con));
			$value['bev']->write($data);
		}
	}
	/**
	 * 告诉前端开启心跳检测
	 */
	private function openHeartbeats ($bev)
	{
		$data = ['type' => 'heartbeats', 'content' => 'open'];
		$data = $this->build(json_encode($data));
		$bev->write($data);
	}


	/**
	 * 定时检查所有用户在线状态，若有人下线发送全员通知
	 */
	private function checkOnlineStatus ()
	{	
		$this->timer(function () {
			foreach ($this->clients as $key => $value) {
				if (!empty($value['last_heartbeats_time'])) {
					$last = $value['last_heartbeats_time'];
					$now = time();
					if (($now - $last) > 10) {
						call_user_func_array(array($this, 'offline'),[$value['fd'], $value['user_id']]);
					}
				}
			}
		}, 10);
	}
	/**
	 * 下线通知
	 */
	private function offline ($fd, $user_id)
	{
		$time = date("Y-m-d H:i:s");
		foreach ($this->clients as $key => $value) {
			if ($value['fd'] == $fd) {
				continue;
			}
			$con = ['type' => 'offline', 'content' => '用户'.$user_id.' '.$time. ' '.'下线了'];
			$data = $this->build(json_encode($con));
			$value['bev']->write($data);
		}
		$this->clients[(int)$fd]['bev']->close();
		unset($this->clients[(int)$fd]);
	}
	/**
	 * 在线人数
	 */
	private function onlineCount ()
	{
		$count = count($this->clients);
		return $count;
	}
	/**
	 * 单独发送信息
	 *
	 */
	private function onlySend ($bev,$con)
	{
		$data = $this->build(json_encode($con));
		$bev->write($data);
	}
	/**
	 * 开启定时器
	 * $callback callable 回调函数
	 * $m int 秒数，间隔几秒运行
	 * $type string 定时器类型 setInterval(按照指定的周期（以秒计）来调用函数或计算表达式。方法会不停地调用函数)
	 * $type string 定时器类型 setTimeout 在指定的秒数后调用函数或计算表达式
	 */
	private function timer (callable $callback , $m = 5, $type = 'setInterval')
	{
		$key = uniqid();
		// $this->timeEvent[$key] = Event::timer($this->base, function($m) use ($callback, $type, $key) {
		//     $return = $callback();
		//     if ($type == 'setInterval') {
		//     	$this->timeEvent[$key]->addTimer($m);
		//     }
		//     if ($return == 'clear' || $type == 'setTimeout') {
		//     	$this->timeEvent[$key]->delTimer();
		//     }
		// }, $m);
		// $this->timeEvent[$key]->addTimer($m);
		$this->ev = Event::timer($this->base, function($m) use ($callback, $type, $key) {
		    $return = $callback();
		    if ($type == 'setInterval') {
		    	$this->ev->addTimer($m);
		    }
		    if ($return == 'clear' || $type == 'setTimeout') {
		    	$this->ev->delTimer();
		    }
		}, $m);
		$this->ev->addTimer($m);
		// $this->ev = $ev;
		return $key;
	}
	
	/**
	 * 事件改变回调
	 * object $bev EventBufferEvent object
	 * $events 事件的位掩码 请参阅 EventBufferEvent常量
	 * $arg 通过EventBufferEvent :: __ construct（） 或 EventBufferEvent :: setCallbacks（）附加到所有回调的自定义变量
	 */
	public function event_callback ($bev, $events, $fd)
	{
		if ($events == $bev::ERROR) $this->error();
		if ($events == $bev::EOF) $bev->free();
	}

	/**
     * 用公共握手算法握手
     *
     * @param $socket
     * @param $buffer
     *
     * @return bool
     */
    public function handShake($val, $buffer, $bev = null) {
        // 获取到客户端的升级密匙
        $line_with_key = substr($buffer, strpos($buffer, 'Sec-WebSocket-Key:') + 18);
        $key = trim(substr($line_with_key, 0, strpos($line_with_key, "\r\n")));

        // 生成升级密匙,并拼接websocket升级头
        $upgrade_key = base64_encode(sha1($key . "258EAFA5-E914-47DA-95CA-C5AB0DC85B11", true));// 升级key的算法
        $upgrade_message = "HTTP/1.1 101 Switching Protocols\r\n";
        $upgrade_message .= "Upgrade: websocket\r\n";
        $upgrade_message .= "Sec-WebSocket-Version: 13\r\n";
        $upgrade_message .= "Connection: Upgrade\r\n";
        $upgrade_message .= "Sec-WebSocket-Accept:" . $upgrade_key . "\r\n\r\n";

        $bev->write($upgrade_message);

        $this->clients[(int)$val] = [
        	'bev'		=> $bev,
			'fd'		=> $val,
			'user_id'	=> '',
			'handShake'	=> true, //是否握手
		];
    }

    /**
     * 解析数据
     *
     * @param $buffer
     *
     * @return bool|string
     */
    private function parse($buffer) {
        $decoded = '';
        $len = ord($buffer[1]) & 127;
        if ($len === 126) {
            $masks = substr($buffer, 4, 4);
            $data = substr($buffer, 8);
        } else if ($len === 127) {
            $masks = substr($buffer, 10, 4);
            $data = substr($buffer, 14);
        } else {
            $masks = substr($buffer, 2, 4);
            $data = substr($buffer, 6);
        }
        for ($index = 0; $index < strlen($data); $index++) {
            $decoded .= $data[$index] ^ $masks[$index % 4];
        }

        return json_decode($decoded, true);
    }

    /**
     * 将普通信息组装成websocket数据帧
     *
     * @param $msg
     *
     * @return string
     */
    private function build($msg) {
        $frame = [];
        $frame[0] = '81';
        $len = strlen($msg);
        if ($len < 126) {
            $frame[1] = $len < 16 ? '0' . dechex($len) : dechex($len);
        } else if ($len < 65025) {
            $s = dechex($len);
            $frame[1] = '7e' . str_repeat('0', 4 - strlen($s)) . $s;
        } else {
            $s = dechex($len);
            $frame[1] = '7f' . str_repeat('0', 16 - strlen($s)) . $s;
        }

        $data = '';
        $l = strlen($msg);
        for ($i = 0; $i < $l; $i++) {
            $data .= dechex(ord($msg{$i}));
        }
        $frame[2] = $data;

        $data = implode('', $frame);

        return pack("H*", $data);
    }
    /**
     * 错误日志记录 
     * $text  string  错误内容
     * $is_cmd_output int  是否在控制台输出错误  1是 2否
     */
	public function error ($text = NULL, $is_cmd_output = 2)
	{
		$file = "/var/www/html/socket/socket/app/log/debug.log";
		if (empty($text)) {
			$text = EventUtil::getLastSocketErrno().'-'.EventUtil::getLastSocketError();
		}
		// $text = date("Y-m-d H:i:s").'--'.$text;
		if ($is_cmd_output == 2) {
			(new Log)->error($text);
		}else {
			echo $text.PHP_EOL;
		}
	}

}
new Server();